package com.wugui.datax.rpc.remoting.net.common;

import cn.hutool.core.collection.CollUtil;
import com.wugui.datax.rpc.remoting.invoker.XxlRpcInvokerFactory;
import com.wugui.datax.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
import com.wugui.datax.rpc.remoting.net.params.BaseCallback;
import com.wugui.datax.rpc.remoting.net.params.XxlRpcRequest;
import com.wugui.datax.rpc.serialize.Serializer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author xuxueli 2018-10-19
 */
public abstract class ConnectClient {

  // ---------------------- iface ----------------------

  private static final ConcurrentMap<String, ConnectClient> CONNECT_CLIENT_MAP =
      new ConcurrentHashMap<>(); // (static) already addStopCallBack
  private static final ConcurrentMap<String, Object> CONNECT_CLIENT_LOCK_MAP =
      new ConcurrentHashMap<>();

  /** async send */
  public static void asyncSend(
      XxlRpcRequest xxlRpcRequest,
      String address,
      Class<? extends ConnectClient> connectClientImpl,
      final XxlRpcReferenceBean xxlRpcReferenceBean)
      throws Exception {

    // client pool	[tips03 : may save 35ms/100invoke if move it to constructor, but it is necessary.
    // cause by ConcurrentHashMap.get]
    ConnectClient clientPool =
        ConnectClient.getPool(address, connectClientImpl, xxlRpcReferenceBean);

    // do invoke
    clientPool.send(xxlRpcRequest);
  }

  private static ConnectClient getPool(
      String address,
      Class<? extends ConnectClient> connectClientImpl,
      final XxlRpcReferenceBean xxlRpcReferenceBean)
      throws Exception {

    // stop callback
    xxlRpcReferenceBean
        .getInvokerFactory()
        .addStopCallBack(
            new BaseCallback() {
              @Override
              public void run() {
                if (CollUtil.isNotEmpty(CONNECT_CLIENT_MAP)) {
                  for (String key : CONNECT_CLIENT_MAP.keySet()) {
                    ConnectClient clientPool = CONNECT_CLIENT_MAP.get(key);
                    clientPool.close();
                  }
                  CONNECT_CLIENT_MAP.clear();
                }
              }
            });

    // get-valid client
    ConnectClient connectClient = CONNECT_CLIENT_MAP.get(address);
    if (connectClient != null && connectClient.isValidate()) {
      return connectClient;
    }

    // lock
    Object clientLock = CONNECT_CLIENT_LOCK_MAP.get(address);
    if (clientLock == null) {
      CONNECT_CLIENT_LOCK_MAP.putIfAbsent(address, new Object());
      clientLock = CONNECT_CLIENT_LOCK_MAP.get(address);
    }

    // remove-create new client
    synchronized (clientLock) {
      // get-valid client, avlid repeat
      connectClient = CONNECT_CLIENT_MAP.get(address);
      if (connectClient != null && connectClient.isValidate()) {
        return connectClient;
      }

      // remove old
      if (connectClient != null) {
        connectClient.close();
        CONNECT_CLIENT_MAP.remove(address);
      }

      // set pool
      ConnectClient connectClient_new = connectClientImpl.newInstance();
      try {
        connectClient_new.init(
            address,
            xxlRpcReferenceBean.getSerializerInstance(),
            xxlRpcReferenceBean.getInvokerFactory());
        CONNECT_CLIENT_MAP.put(address, connectClient_new);
      } catch (Exception e) {
        connectClient_new.close();
        throw e;
      }

      return connectClient_new;
    }
  }

  // ---------------------- client pool map ----------------------

  public abstract void init(
      String address, final Serializer serializer, final XxlRpcInvokerFactory xxlRpcInvokerFactory)
      throws Exception;

  public abstract void close();

  public abstract boolean isValidate();

  public abstract void send(XxlRpcRequest xxlRpcRequest) throws Exception;
}
